package org.ala.distributed_transaction_commons.communication.participant;

import java.util.concurrent.atomic.AtomicInteger;

import org.ala.distributed_transaction_commons.communication.Message;
import org.ala.distributed_transaction_commons.communication.Message.MessageContent;
import org.ala.distributed_transaction_commons.communication.coordinator.CoordinatorSubscriber.ICallable;
import org.ala.distributed_transaction_commons.logs.LoggerFactory;
import org.ala.distributed_transaction_commons.redis.DistributedTransactionRedisTools;
import org.slf4j.Logger;
import org.springframework.beans.factory.InitializingBean;

import redis.clients.jedis.JedisPubSub;


/**
 * 参与者的scriber
 *
 * @author 骆毅(ala)
 * @date 2024年2月20日
 */
public class ParticipantSubcriber implements InitializingBean {
	
	
	static Logger log = LoggerFactory.getLogger("distributed_transaction/participant", "subscriber");
	
	
	/**
	 * redis工具
	 */
	protected DistributedTransactionRedisTools distributedTransactionRedisTools;
	/**
	 * 监听到消息回调
	 */
	protected ICallable<MessageContent> onMessage;
	/**
	 * 开始监听回调
	 */
	protected ICallable<String> onSubscrib;
	/**
	 * 监听撤销回调
	 */
	protected ICallable<String> onUnSubscrib;
	

	@Override
	public void afterPropertiesSet() throws Exception {
		
	}
	/**
	 * 监听线程计数器
	 */
	protected AtomicInteger subscribThreadNum = new AtomicInteger(0);
	/**
	 * 监听全部发送给协调者的channel
	 */
	public void reSubscrib(String[] channels) {
		new Thread("ParticipantSubcriberThread_" + subscribThreadNum.getAndIncrement()) {
			public void run() {
				distributedTransactionRedisTools.withJedis(jedis -> {
					jedis.subscribe(new JedisPubSub() {
						public void onMessage(String channel, String message) {
							ParticipantSubcriber.this.onMessage(channel, message);
						}
						public void onSubscribe(String channel, int subscribedChannels) {
							if (onSubscrib != null) {onSubscrib.call(channel);}
						}
						public void onUnsubscribe(String channel, int subscribedChannels) {
							if (onUnSubscrib != null) {onUnSubscrib.call(channel);}
						}
					}, channels);
					return null;
				});
			}
		}.start();
	}
	
	
	/**
	 * 收到消息回调
	 */
	protected void onMessage(String channel, String msg) {
		//	解析message
		MessageContent mc = Message.parseMessage(msg);
		if (mc == null) {
			log.warn("[ParticipantSubcriber onMessage] 监听到消息，但消息无法解析 msg:" + msg);
			return;
		}
		
		if (onMessage != null) {
			onMessage.call(mc);
		}
	}


	public DistributedTransactionRedisTools getDistributedTransactionRedisTools() {
		return distributedTransactionRedisTools;
	}
	public void setDistributedTransactionRedisTools(DistributedTransactionRedisTools distributedTransactionRedisTools) {
		this.distributedTransactionRedisTools = distributedTransactionRedisTools;
	}
	public ICallable<MessageContent> getOnMessage() {
		return onMessage;
	}
	public void setOnMessage(ICallable<MessageContent> onMessage) {
		this.onMessage = onMessage;
	}
	public ICallable<String> getOnSubscrib() {
		return onSubscrib;
	}
	public void setOnSubscrib(ICallable<String> onSubscrib) {
		this.onSubscrib = onSubscrib;
	}
	public ICallable<String> getOnUnSubscrib() {
		return onUnSubscrib;
	}
	public void setOnUnSubscrib(ICallable<String> onUnSubscrib) {
		this.onUnSubscrib = onUnSubscrib;
	}
	
}
